package org.apache.rocketmq.example.MessageTestAll.Produce;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.example.quickstart.Producer;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;

public class quikstart_P {

public void  Sync() throws MQClientException, UnsupportedEncodingException, MQBrokerException, RemotingException, InterruptedException {

    DefaultMQProducer defaultMQProducer = new DefaultMQProducer(Producer.PRODUCER_GROUP);
    defaultMQProducer.setNamesrvAddr(Producer.DEFAULT_NAMESRVADDR);

defaultMQProducer.setSendMsgTimeout(10000);
defaultMQProducer.start();

    System.out.println("同步消息生产者启动成功");

    // 发送10条同步消息
    for (int i = 0; i < 10; i++) {
        // 创建消息，指定主题、标签和消息体
        Message message = new Message("sync_topic",
                "sync_tag",
                ("同步消息示例-" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
        // 发送消息并等待响应
        SendResult sendResult = defaultMQProducer.send(message);

        // 打印发送结果
        System.out.printf("消息发送结果: %s, 消息ID: %s, 队列: %s%n",
                sendResult.getSendStatus(),
                sendResult.getMsgId(),
                sendResult.getMessageQueue().getQueueId());
    }

    // 关闭生产者
    defaultMQProducer.shutdown();
    System.out.println("同步消息生产者已关闭");



}


public void Async()throws MQClientException, UnsupportedEncodingException, MQBrokerException, RemotingException, InterruptedException {
    DefaultMQProducer defaultMQProducer = new DefaultMQProducer(Producer.PRODUCER_GROUP);
    defaultMQProducer.setNamesrvAddr(Producer.DEFAULT_NAMESRVADDR);

    defaultMQProducer.setSendMsgTimeout(10000);

    defaultMQProducer.start();

    System.out.println("同步消息生产者启动成功");

    // 发送10条步消息
    for (int i = 0; i < 10; i++) {
        // 创建消息，指定主题、标签和消息体
        Message message = new Message("Async_topic",
                "Async_tag",
                ("异步消息示例-" + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
        // 发送消息并等待响应
        defaultMQProducer.send(message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                // 打印发送结果
                System.out.printf("消息发送结果: %s, 消息ID: %s, 队列: %s%n",
                        sendResult.getSendStatus(),
                        sendResult.getMsgId(),
                        sendResult.getMessageQueue().getQueueId());
            }

            @Override
            public void onException(Throwable e) {
                e.printStackTrace();
                String s =  this.getClass().getName()+ "----" + e.getClass().getName();
                System.out.println(s);
            }
        });


    }

    Thread.sleep(5000);
    // 关闭生产者
    defaultMQProducer.shutdown();
    System.out.println("同步消息生产者已关闭");



}

public void Order() throws MQClientException, UnsupportedEncodingException, MQBrokerException, RemotingException, InterruptedException {
    // 创建生产者
    DefaultMQProducer producer = new DefaultMQProducer("orderly_producer_group");
    producer.setNamesrvAddr("localhost:9876");
    producer.start();

    System.out.println("顺序消息生产者启动成功");

    // 定义订单ID列表，用于确保同一订单的消息发送到同一队列
    String[] orderIds = {"order_1001", "order_1002", "order_1003", "order_1001", "order_1002"};

    // 发送顺序消息
    for (int i = 0; i < orderIds.length; i++) {
        String orderId = orderIds[i];
        Message message = new Message("orderly_topic",
                "orderly_tag",
                ("订单消息-" + i + "-" + orderId).getBytes(RemotingHelper.DEFAULT_CHARSET));

        // 自定义队列选择器，确保同一订单ID的消息发送到同一队列
        producer.send(message, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                String id = (String) arg;
                // 通过订单ID哈希值选择队列
                int index = id.hashCode() % mqs.size();
                return mqs.get(index);
            }
        }, orderId); // 传递订单ID作为参数

        System.out.printf("发送顺序消息: 订单ID=%s, 消息内容=%s%n",
                orderId,
                new String(message.getBody(), RemotingHelper.DEFAULT_CHARSET));
    }

    producer.shutdown();
    System.out.println("顺序消息生产者已关闭");
}

public void Delay()  throws MQClientException, UnsupportedEncodingException, MQBrokerException, RemotingException, InterruptedException{
    // 创建生产者
    DefaultMQProducer producer = new DefaultMQProducer("delay_producer_group");
    producer.setNamesrvAddr("localhost:9876");
    producer.start();

    System.out.println("延迟消息生产者启动成功");

    // 定义延迟级别（1s, 5s, 10s, 30s, 1m, 2m, 3m, 4m, 5m, 6m, 7m, 8m, 9m, 10m, 20m, 30m, 1h, 2h）
    int[] delayLevels = {10}; // 分别对应1s, 5s, 10s, 30s, 2m

    // 发送延迟消息
    for (int i = 0; i < delayLevels.length; i++) {
        Message message = new Message("delay_topic",
                "delay_tag",
                ("延迟消息示例-" + i + "-" + new Date()).getBytes(RemotingHelper.DEFAULT_CHARSET));

        // 设置延迟级别
        message.setDelayTimeLevel(delayLevels[i]);

        // 发送消息
        SendResult sendResult = producer.send(message);

        System.out.printf("发送延迟消息: 级别=%d, 状态=%s, 消息ID=%s, 内容=%s%n",
                delayLevels[i],
                sendResult.getSendStatus(),
                sendResult.getMsgId(),
                new String(message.getBody(), RemotingHelper.DEFAULT_CHARSET));
    }

    producer.shutdown();
    System.out.println("延迟消息生产者已关闭");
}


    public static void main(String[] args) {
        quikstart_P quikstartP = new quikstart_P();
        Method[] methods = quikstart_P.class.getMethods();
        for (int i = 0; i < methods.length; i++) {

           if(methods[i].getDeclaringClass() == Object.class)
               continue;

            try {
                if(methods[i].getName().equals("main"))
                {
                    continue;
                }
                methods[i].invoke(quikstartP);
            } catch (IllegalAccessException e) {
                throw new RuntimeException(e);
            } catch (InvocationTargetException e) {
                throw new RuntimeException(e);
            }
        }

while (true);
    }

}



